package com.ccubee.data.transfer.transfer;

import cn.hutool.core.collection.CollUtil;
import com.ccubee.data.transfer.model.TransferRequest;
import com.ccubee.data.transfer.properties.DataProperties;
import com.ccubee.data.transfer.properties.transfer.DbProperties;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;

/**
 * 数据传输处理
 *
 * @Author ycf
 * @Date 2022/11/13 22:41
 * @Version 1.0
 */
@Component
@Slf4j
public class TransferImpl implements ITransfer {
    @Resource
    private DataProperties dataProperties;
    private final static Map<String, ICall> I_CALL_MAP = new ConcurrentHashMap<>(16);

    @SneakyThrows
    @Override
    public Set<String> start(TransferRequest transferRequest) {
        Set<String> instances = Set.of(transferRequest.getInstance().split(","));
        return dataProperties.getTransfer()
                .getInstance()
                .entrySet()
                .stream()
                .filter(i -> CollUtil.isNotEmpty(instances) && instances.contains(i.getKey()))
                .filter(i -> cd(i, transferRequest))
                .map(Map.Entry::getKey)
                .collect(Collectors.toSet());
    }

    /**
     * 执行任务
     *
     * @param v               数据
     * @param transferRequest 请求数据
     */
    @SneakyThrows
    private boolean cd(Map.Entry<String, List<DbProperties>> v, TransferRequest transferRequest) {
        if (I_CALL_MAP.containsKey(v.getKey()) && I_CALL_MAP.get(v.getKey()).isRun()) {
            log.warn("task job:{} is run but request task stop!,{}", v.getKey(), transferRequest);
            return false;
        }
        DataProperties.InstanceThreadPool instanceThreadPool = dataProperties.getThreadPool(v.getKey());
        ICall iCall = TransferContext.get(transferRequest.getTransferMod())
                .getClass()
                .getConstructor()
                .newInstance()
                .init(v.getKey(), dataProperties.getTransfer().getInstance().get(v.getKey()),
                        dataProperties, new ThreadPoolExecutor(instanceThreadPool.getCorePoolSize(),
                                instanceThreadPool.getMaximumPoolSize(), instanceThreadPool.getKeepAliveTime(),
                                instanceThreadPool.getUnit(),
                                new LinkedBlockingQueue<>(instanceThreadPool.getBlockingQueue()),
                                new ThreadFactoryBuilder()
                                        .setNameFormat(v.getKey() + "-%d").build(), instanceThreadPool.getRejectedType().rejected()),
                        new ScheduledThreadPoolExecutor(1, new com.alibaba.google.common.util.concurrent.ThreadFactoryBuilder()
                                .setNameFormat(v.getKey() + "-Scheduled" + "-%d").build(), instanceThreadPool.getRejectedType().rejected()))
                .exe(transferRequest);
        iCall.start();
        I_CALL_MAP.put(v.getKey(), iCall);
        return true;
    }

    @Override
    public void stop(Set<String> key) {
        if (I_CALL_MAP.isEmpty()) {
            log.warn("job not run");
            return;
        }
        if (CollUtil.isEmpty(key)) {
            key = I_CALL_MAP.keySet();
        }
        key.forEach(i -> {
            if (I_CALL_MAP.containsKey(i)) {
                I_CALL_MAP.get(i).stop();
                log.info("job end");
            } else {
                log.warn("job not contains:{}", i);
            }

        });


    }
}
